package com.leyantech.ecp.agenx.dispatcher.client;

import com.leyantech.utility.config.ConfigUtil;

import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-12-05.
 */
@Slf4j
@Singleton
public class AgenxDispatcher {

  private final Producer<String, String> producer;

  private final Supplier<String> messageTopic;

  @Inject
  public AgenxDispatcher(@Named("dispatch-client") Producer<String, String> producer) {
    final Supplier<String> messageTopic = ConfigUtil.getDynamicConfig(
        "1.ecp-agenx-pdd",
        "dispatcher.message.topic",
        Function.identity(),
        ""
    );
    Preconditions.checkArgument(StringUtils.isNotBlank(messageTopic.get()), "topic is blank.");
    this.messageTopic = messageTopic;

    this.producer = producer;
  }


  /**
   * 发送给 connector 上的所有在线客服.
   */
  public void sendToAll(String message) {
    send("all", message);
  }

  /**
   * 发送给 connector 上的指定店铺的所有在线客服.
   */
  public void sendToStoreConnector(String sid, String message) {
    send(sid, message);
  }

  /**
   * 发送给 connector 上的指定店铺的在线客服.
   */
  public void sendToConnector(String sid, String assistantId, String message) {
    send(String.join(".", sid, assistantId), message);
  }

  public void send(String address, String message) {
    final String topic = messageTopic.get();
    final ProducerRecord<String, String> record = new ProducerRecord<>(topic, address, message);
    producer.send(record, (metadata, exception) -> {
      if (exception != null) {
        log.error("send failed, {} {} {} ", topic, address, message, exception);
      } else {
        log.debug("send succeed, {} {} {} ", topic, address, message);
      }
    });
  }

}
